package executor

import (
	"database/sql"
	"fmt"
	"sort"
	"strings"
	"sync"

	"kwdb-import/pkg/parser"

	_ "github.com/lib/pq"
)

// StatementExecutor SQL语句执行器
type StatementExecutor struct {
	db *sql.DB
}

// NewStatementExecutor 创建新的执行器
func NewStatementExecutor(connectionString string) (*StatementExecutor, error) {
	db, err := sql.Open("postgres", connectionString)
	if err != nil {
		return nil, fmt.Errorf("连接数据库失败: %v", err)
	}

	// 测试连接
	if err := db.Ping(); err != nil {
		return nil, fmt.Errorf("数据库连接测试失败: %v", err)
	}

	return &StatementExecutor{db: db}, nil
}

// Close 关闭数据库连接
func (e *StatementExecutor) Close() error {
	if e.db != nil {
		return e.db.Close()
	}
	return nil
}

// ExecuteStatements 执行SQL语句列表
func (e *StatementExecutor) ExecuteStatements(statements []string) error {
	// 开始事务
	tx, err := e.db.Begin()
	if err != nil {
		return fmt.Errorf("开始事务失败: %v", err)
	}

	defer func() {
		if err != nil {
			tx.Rollback()
		}
	}()

	// 执行每个语句
	for i, stmt := range statements {
		if strings.TrimSpace(stmt) == "" {
			continue
		}

		fmt.Printf("执行语句 %d/%d: %s\n", i+1, len(statements), e.truncateSQL(stmt))

		if _, err := tx.Exec(stmt); err != nil {
			return fmt.Errorf("执行语句失败 (第%d条): %v\nSQL: %s", i+1, err, stmt)
		}
	}

	// 提交事务
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("提交事务失败: %v", err)
	}

	return nil
}

// ExecuteStatementsWithOrder 按优先级顺序执行语句
func (e *StatementExecutor) ExecuteStatementsWithOrder(statements []parser.SQLStatement, convertedStatements []string) error {
	// 创建语句和转换后语句的映射
	stmtMap := make(map[int]string)
	for i, converted := range convertedStatements {
		if i < len(statements) {
			stmtMap[i] = converted
		}
	}

	// 按优先级排序
	sortedIndices := e.sortStatementsByPriority(statements)

	// 按顺序执行
	var orderedStatements []string
	for _, index := range sortedIndices {
		if converted, exists := stmtMap[index]; exists && strings.TrimSpace(converted) != "" {
			orderedStatements = append(orderedStatements, converted)
		}
	}

	return e.ExecuteStatements(orderedStatements)
}

// sortStatementsByPriority 按优先级排序语句
func (e *StatementExecutor) sortStatementsByPriority(statements []parser.SQLStatement) []int {
	// 定义执行优先级
	priorityMap := map[parser.StatementType]int{
		parser.UseDatabase:     0,
		parser.CreateDatabase:  1,
		parser.DropDatabase:    2,
		parser.CreateTable:     3,
		parser.DropTable:       4,
		parser.CreateIndex:     5,
		parser.CreateSequence:  6,
		parser.CreateView:      7,
		parser.Insert:          8,
		parser.CreateProcedure: 9,
		parser.CreateFunction:  9,
		parser.Delimiter:       10,
		parser.CreateTrigger:   11,
		parser.Unknown:         99,
	}

	// 创建索引和优先级的映射
	type indexPriority struct {
		index    int
		priority int
		stmtType parser.StatementType
	}

	var items []indexPriority
	for i, stmt := range statements {
		priority, exists := priorityMap[stmt.Type]
		if !exists {
			priority = 99 // 未知类型的优先级最低
		}
		items = append(items, indexPriority{
			index:    i,
			priority: priority,
			stmtType: stmt.Type,
		})
	}

	// 按优先级排序
	sort.Slice(items, func(i, j int) bool {
		if items[i].priority == items[j].priority {
			// 相同优先级按原始顺序
			return items[i].index < items[j].index
		}
		return items[i].priority < items[j].priority
	})

	// 提取排序后的索引
	var sortedIndices []int
	for _, item := range items {
		sortedIndices = append(sortedIndices, item.index)
	}

	return sortedIndices
}

// truncateSQL 截断SQL语句用于显示
func (e *StatementExecutor) truncateSQL(sql string) string {
	const maxLength = 100
	if len(sql) <= maxLength {
		return sql
	}
	return sql[:maxLength] + "..."
}

// ValidateConnection 验证数据库连接
func (e *StatementExecutor) ValidateConnection() error {
	if e.db == nil {
		return fmt.Errorf("数据库连接未初始化")
	}
	return e.db.Ping()
}

// GetDatabaseInfo 获取数据库信息
func (e *StatementExecutor) GetDatabaseInfo() (map[string]interface{}, error) {
	info := make(map[string]interface{})

	// 获取数据库版本
	var version string
	err := e.db.QueryRow("SELECT version()").Scan(&version)
	if err != nil {
		return nil, fmt.Errorf("获取数据库版本失败: %v", err)
	}
	info["version"] = version

	// 获取当前数据库名
	var dbName string
	err = e.db.QueryRow("SELECT current_database()").Scan(&dbName)
	if err != nil {
		return nil, fmt.Errorf("获取当前数据库名失败: %v", err)
	}
	info["database"] = dbName

	// 获取当前用户
	var user string
	err = e.db.QueryRow("SELECT current_user").Scan(&user)
	if err != nil {
		return nil, fmt.Errorf("获取当前用户失败: %v", err)
	}
	info["user"] = user

	return info, nil
}

// ExecuteSingleStatement 执行单个语句
func (e *StatementExecutor) ExecuteSingleStatement(sql string) error {
	if strings.TrimSpace(sql) == "" {
		return nil
	}

	_, err := e.db.Exec(sql)
	if err != nil {
		return fmt.Errorf("执行语句失败: %v\nSQL: %s", err, sql)
	}

	return nil
}

// BeginTransaction 开始事务
func (e *StatementExecutor) BeginTransaction() (*sql.Tx, error) {
	return e.db.Begin()
}

// ExecuteStatementsParallel 并发执行SQL语句列表（支持外键重试）
// workerCount: 并发worker数量
// batchSize: 每个批次的语句数量
// onError: 错误处理策略 ("continue" 或 "stop")
// maxErrors: 最大允许错误数
// 返回: 成功数, 错误数, 外键失败列表, 错误
func (e *StatementExecutor) ExecuteStatementsParallel(statements []string, workerCount int, batchSize int, onError string, maxErrors int) (executedCount int, errorCount int, fkFailures []string, err error) {
	if len(statements) == 0 {
		return 0, 0, nil, nil
	}

	// 将语句分批
	batches := make([][]string, 0)
	for i := 0; i < len(statements); i += batchSize {
		end := i + batchSize
		if end > len(statements) {
			end = len(statements)
		}
		batches = append(batches, statements[i:end])
	}

	// 创建工作通道
	type batchJob struct {
		index int
		stmts []string
	}
	jobChan := make(chan batchJob, len(batches))

	// 创建结果收集通道
	type batchResult struct {
		index      int
		success    int
		errors     int
		fkErrors   []string // 外键约束失败的语句
		firstError error
	}
	resultChan := make(chan batchResult, len(batches))

	// 启动worker池
	for w := 0; w < workerCount; w++ {
		go func(workerID int) {
			for job := range jobChan {
				successCount := 0
				errCount := 0
				fkErrStmts := make([]string, 0)
				var firstErr error

				for _, stmt := range job.stmts {
					trimmed := strings.TrimSpace(stmt)
					if trimmed == "" {
						continue
					}

					if err := e.ExecuteSingleStatement(trimmed); err != nil {
						// 检查是否为外键约束错误
						if isForeignKeyError(err) {
							fkErrStmts = append(fkErrStmts, trimmed)
						} else {
							if firstErr == nil {
								firstErr = err
							}
							errCount++
						}
					} else {
						successCount++
					}
				}

				resultChan <- batchResult{
					index:      job.index,
					success:    successCount,
					errors:     errCount,
					fkErrors:   fkErrStmts,
					firstError: firstErr,
				}
			}
		}(w)
	}

	// 发送所有任务
	for i, batch := range batches {
		jobChan <- batchJob{index: i, stmts: batch}
	}
	close(jobChan)

	// 收集结果
	totalSuccess := 0
	totalErrors := 0
	allFkFailures := make([]string, 0)
	var collectedError error

	for i := 0; i < len(batches); i++ {
		result := <-resultChan
		totalSuccess += result.success
		totalErrors += result.errors
		allFkFailures = append(allFkFailures, result.fkErrors...)

		if result.firstError != nil && collectedError == nil {
			collectedError = result.firstError
		}

		// 检查是否超过最大错误数
		if totalErrors >= maxErrors {
			if onError == "stop" {
				return totalSuccess, totalErrors, allFkFailures, fmt.Errorf("错误数量超过限制 (%d): %v", maxErrors, collectedError)
			}
		}
	}

	close(resultChan)

	if totalErrors > 0 && onError == "stop" && collectedError != nil {
		return totalSuccess, totalErrors, allFkFailures, collectedError
	}

	return totalSuccess, totalErrors, allFkFailures, nil
}

// ExecuteStatementsParallelStream 并发流式执行：从通道消费语句，降低内存占用
// workerCount: 并发worker数量
// onError: 错误处理策略 ("continue" 或 "stop")
// maxErrors: 最大允许错误数
// 返回: 成功数, 错误数, 外键失败列表, 错误
func (e *StatementExecutor) ExecuteStatementsParallelStream(dmlChan <-chan string, workerCount int, onError string, maxErrors int) (executedCount int, errorCount int, fkFailures []string, err error) {
	if workerCount <= 0 {
		workerCount = 1
	}

	type workerResult struct {
		success    int
		errors     int
		fkErrors   []string
		firstError error
	}

	resultChan := make(chan workerResult, workerCount)
	var wg sync.WaitGroup

	// 启动worker池，直接从dmlChan消费并执行
	for w := 0; w < workerCount; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			successCount := 0
			errCount := 0
			fkErrStmts := make([]string, 0)
			var firstErr error

			for stmt := range dmlChan {
				trimmed := strings.TrimSpace(stmt)
				if trimmed == "" {
					continue
				}

				if err := e.ExecuteSingleStatement(trimmed); err != nil {
					if isForeignKeyError(err) {
						fkErrStmts = append(fkErrStmts, trimmed)
					} else {
						if firstErr == nil {
							firstErr = err
						}
						errCount++
					}
				} else {
					successCount++
				}
			}

			resultChan <- workerResult{
				success:    successCount,
				errors:     errCount,
				fkErrors:   fkErrStmts,
				firstError: firstErr,
			}
		}()
	}

	// 等待所有worker结束后关闭结果通道
	go func() {
		wg.Wait()
		close(resultChan)
	}()

	totalSuccess := 0
	totalErrors := 0
	allFkFailures := make([]string, 0)
	var collectedError error

	for res := range resultChan {
		totalSuccess += res.success
		totalErrors += res.errors
		allFkFailures = append(allFkFailures, res.fkErrors...)

		if res.firstError != nil && collectedError == nil {
			collectedError = res.firstError
		}

		if totalErrors >= maxErrors && onError == "stop" && collectedError != nil {
			return totalSuccess, totalErrors, allFkFailures, fmt.Errorf("错误数量超过限制 (%d): %v", maxErrors, collectedError)
		}
	}

	if totalErrors > 0 && onError == "stop" && collectedError != nil {
		return totalSuccess, totalErrors, allFkFailures, collectedError
	}

	return totalSuccess, totalErrors, allFkFailures, nil
}

// isForeignKeyError 检查是否为外键约束错误
func isForeignKeyError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "sqlstate 23503") || strings.Contains(msg, "violates foreign key constraint")
}
